package f5transformation.aggregation;

import flinkemp.Emp;
import flinkemp.EmpFun;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class KeyByReduce {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Emp> data = env.addSource(new EmpFun());
        data.print();

        //转换数据格式
        SingleOutputStreamOperator<Tuple3<Integer, Double, Integer>> map = data.map(new MapFunction<Emp, Tuple3<Integer, Double, Integer>>() {
            @Override
            public Tuple3<Integer, Double, Integer> map(Emp emp) throws Exception {
                return Tuple3.of(emp.deptNo, emp.sal, 1);
            }
        });

        //分组操作
        KeyedStream<Tuple3<Integer, Double, Integer>, Integer> tuple3IntegerKeyedStream = map.keyBy(new KeySelector<Tuple3<Integer, Double, Integer>, Integer>() {
            @Override
            public Integer getKey(Tuple3<Integer, Double, Integer> integerDoubleIntegerTuple3) throws Exception {
                return integerDoubleIntegerTuple3.f0;
            }
        });
        //聚合操作
        SingleOutputStreamOperator<Tuple3<Integer, Double, Integer>> reduce = tuple3IntegerKeyedStream.reduce(new ReduceFunction<Tuple3<Integer, Double, Integer>>() {
            @Override
            public Tuple3<Integer, Double, Integer> reduce(Tuple3<Integer, Double, Integer> integerDoubleIntegerTuple3, Tuple3<Integer, Double, Integer> t1) throws Exception {
                Integer deptNo = integerDoubleIntegerTuple3.f0;
                //当前累计结果与新纪录相加
                Double salSum = integerDoubleIntegerTuple3.f1 + t1.f1;
                Integer count = integerDoubleIntegerTuple3.f2 + t1.f2;
                return new Tuple3(deptNo, salSum, count);
            }
        });
        //转换结果
        reduce.map(new MapFunction<Tuple3<Integer, Double, Integer>, String>() {
            @Override
            public String map(Tuple3<Integer, Double, Integer> T3) throws Exception {
                return "部门:" + T3.f0 +
                        "\t薪资和:" + T3.f1 +
                        "\t员工数:" + T3.f2 +
                        "\t薪资均值:" + T3.f1 / T3.f2;
            }
        }).print();
        env.execute();
    }
}
